﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using NUnit.Framework;
using UnitTest.Helpers;
using XcoAppSpaces.Core;
using XcoAppSpaces.Core.Ports;
using XcoAppSpaces.Contracts.Exceptions;

namespace test.XcoAppSpaces.Core
{
	[TestFixture]
	public class testXcoPublisher
	{
		private Dictionary<Type, Dictionary<Guid, XcoPublisher.Subscription>> GetSubscriptions(XcoPublisher publisher)
		{
			return publisher.GetPrivateField("subscriptions") as Dictionary<Type, Dictionary<Guid, XcoPublisher.Subscription>>;
		}

		[Test]
		public void TestInstantiate()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Assert.IsTrue(GetSubscriptions(publisher).ContainsKey(typeof(int)));
			Assert.IsTrue(GetSubscriptions(publisher).ContainsKey(typeof(string)));
			Assert.IsTrue(GetSubscriptions(publisher).ContainsKey(typeof(DateTime)));
		}

		[Test]
		public void TestSubscribeLocalOk()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Port<string> port = new Port<string>();
			Subscribe sub = new Subscribe<string>(port);
			publisher.ProcessSubscribe(sub);
			Assert.IsTrue(GetSubscriptions(publisher)[typeof(string)].Values.Any(s => s.Subscriber == sub));
		}

		[Test]
		[ExpectedException(typeof(XcoSubscriptionException))]
		public void TestSubscribeLocalUnknown()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(DateTime));
			Port<string> port = new Port<string>();
			publisher.ProcessSubscribe(new Subscribe<string>(port));
		}

		[Test]
		[ExpectedException(typeof(XcoSubscriptionException))]
		public void TestSubscribeNull()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			publisher.ProcessSubscribe(new Subscribe<string>(null));
		}

		[Test]
		public void TestUnsubscribeLocalOk()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Port<string> port = new Port<string>();
			Subscribe sub = new Subscribe<string>(port);
			publisher.ProcessSubscribe(sub);
			publisher.ProcessUnsubscribe(new Unsubscribe<string>(port));
			Assert.IsFalse(GetSubscriptions(publisher)[typeof(string)].Values.Any(s => s.Subscriber == sub));
		}

		[Test]
		public void TestUnsubscribeLocalUnknown()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Port<string> port = new Port<string>();
			Subscribe sub = new Subscribe<string>(port);
			publisher.ProcessSubscribe(sub);
			Port<string> port2 = new Port<string>();
			publisher.ProcessUnsubscribe(new Unsubscribe<string>(port2)); //should not throw an error
			Assert.IsTrue(GetSubscriptions(publisher)[typeof(string)].Values.Any(s => s.Subscriber == sub));
		}

		[Test]
		public void TestUnsubscribeNull()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			publisher.ProcessUnsubscribe(new Unsubscribe<string>(null)); //should not throw an error
		}

		[Test]
		public void TestPublish()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Port<string> port1 = new Port<string>();
			Port<string> port2 = new Port<string>();
			publisher.ProcessSubscribe(new Subscribe<string>(port1));
			publisher.ProcessSubscribe(new Subscribe<string>(port2));
			string teststr = "test";
			publisher.PublishUnknownType(teststr);

			Thread.Sleep(100);
			
			Assert.AreEqual(1, port1.ItemCount);
			Assert.AreEqual(1, port2.ItemCount);
			Assert.AreEqual(teststr, port1.Receive(1000).Result);
			Assert.AreEqual(teststr, port2.Receive(1000).Result);

			publisher.ProcessUnsubscribe(new Unsubscribe<string>(port1));
			publisher.ProcessUnsubscribe(new Unsubscribe<string>(port2));
			publisher.PublishUnknownType(teststr);
			
			Assert.AreEqual(0, port1.ItemCount);
			Assert.AreEqual(0, port2.ItemCount);
		}

        [Test]
        public void TestSubUnsubWithGenericPort()
        {
            XcoPublisher publisher = new XcoPublisher(typeof(int));
            Port<object> port1 = new Port<object>();
            publisher.ProcessSubscribe(new Subscribe<int>(port1));

            publisher.PublishUnknownType(99);

            Assert.AreEqual(99, port1.Receive(500).Result);

            publisher.ProcessUnsubscribe(new Unsubscribe<int>(port1));
            publisher.PublishUnknownType(100);

            try
            {
                port1.Receive(500).Wait();
                Assert.Fail("There shouldn´t be another element in the port.");
            }
            catch(AggregateException)
            {}
            catch(Exception ex)
            {
                Assert.Fail("Unexpected exception: " + ex.Message);
            }

        }


		[Test]
		[ExpectedException(typeof(XcoPublishException))]
		public void TestPublishUnknown()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(string), typeof(DateTime));
			Port<string> port = new Port<string>();
			publisher.ProcessSubscribe(new Subscribe<string>(port));
			publisher.PublishUnknownType(123);
		}

		[Test]
		public void TestPublishWithFilter()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int), typeof(string), typeof(DateTime));
			Port<int> port1 = new Port<int>();
			Port<int> port2 = new Port<int>();
			publisher.ProcessSubscribe(new Subscribe<int>(port1, num => num % 2 == 0));
			publisher.ProcessSubscribe(new Subscribe<int>(port2, num => num % 2 == 1));
			publisher.PublishUnknownType(1);
			publisher.PublishUnknownType(2);
			publisher.PublishUnknownType(3);
			publisher.PublishUnknownType(4);

			Thread.Sleep(100);
			
			Assert.AreEqual(2, port1.ItemCount);
			Assert.AreEqual(2, port2.ItemCount);
			Assert.AreEqual(2, port1.Receive(1000).Result);
			Assert.AreEqual(4, port1.Receive(1000).Result);
			Assert.AreEqual(1, port2.Receive(1000).Result);
			Assert.AreEqual(3, port2.Receive(1000).Result);
		}

		[Test]
		public void TestPublish_ItemsKeepSequentialOrder()
		{
			XcoPublisher publisher = new XcoPublisher(typeof(int));
			Port<int> port1 = new Port<int>();
			Port<int> port2 = new Port<int>();
			publisher.ProcessSubscribe(new Subscribe<int>(port1));
			publisher.ProcessSubscribe(new Subscribe<int>(port2));

			for (int i = 0; i < 100; i++)
			{
				publisher.PublishUnknownType(i);
			}

			SpinWait.SpinUntil(() => port1.ItemCount == 100 && port2.ItemCount == 100, 1000);

			Assert.AreEqual(100, port1.ItemCount);
			Assert.AreEqual(100, port2.ItemCount);

			for (int i = 0; i < 100; i++)
			{
				Assert.AreEqual(i, port1.Receive(100).Result);
				Assert.AreEqual(i, port2.Receive(100).Result);
			}
		}

		[Test]
		public void TestPublish_Remote_PortRemovedOnError()
		{
			using (XcoAppSpace client = new XcoAppSpace("tcp.port=8001"),
				server = new XcoAppSpace("tcp.port=8002"))
			{
				XcoPublisher publisher = new XcoPublisher(typeof(int));
				server.Run<Subscribe<int>>(publisher.ProcessSubscribe);

				var regPort = client.Connect<Subscribe<int>>("localhost:8002");
				var subPort = new Port<int>();
				regPort.Post(new Subscribe<int>(subPort));

				//test successful remote subscription
				Thread.Sleep(100);
				publisher.PublishUnknownType(123);
				var result = subPort.Receive(500).Result;
				Assert.AreEqual(123, result);
				Assert.AreEqual(1, GetSubscriptions(publisher)[typeof(int)].Count);

				//when client is disposed, no more messages should arrive and the subscription should automatically be removed
				client.Dispose();
				publisher.PublishUnknownType(123);
				Thread.Sleep(2000);
				Assert.AreEqual(0, subPort.ItemCount);
				Assert.AreEqual(0, GetSubscriptions(publisher)[typeof(int)].Count);
			}
		}
	}
}
